package org.swqsbssddqs.shortlink.project.service.impl;


import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jakarta.servlet.ServletRequest;
import jakarta.servlet.ServletResponse;
import jakarta.servlet.http.Cookie;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.swqsbssddqs.shortlink.project.common.convention.exception.ClientException;
import org.swqsbssddqs.shortlink.project.common.convention.exception.ServiceException;
import org.swqsbssddqs.shortlink.project.common.enums.VailDateTypeEnum;
import org.swqsbssddqs.shortlink.project.config.ShortLinkGotoWhiteList;
import org.swqsbssddqs.shortlink.project.dao.entity.ShortLinkDO;
import org.swqsbssddqs.shortlink.project.dao.entity.ShortLinkGotoDO;
import org.swqsbssddqs.shortlink.project.dao.mapper.*;
import org.swqsbssddqs.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import org.swqsbssddqs.shortlink.project.dto.req.ShortLinkBatchCreateReqDTO;
import org.swqsbssddqs.shortlink.project.dto.req.ShortLinkCreatReqDTO;
import org.swqsbssddqs.shortlink.project.dto.req.ShortLinkPageReqDTO;
import org.swqsbssddqs.shortlink.project.dto.req.ShortLinkUpdateReqDTO;
import org.swqsbssddqs.shortlink.project.dto.resp.*;
import org.swqsbssddqs.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import org.swqsbssddqs.shortlink.project.mq.producer.ShortLinkStatsSaveProducer;
import org.swqsbssddqs.shortlink.project.service.ShortLinkService;
import org.swqsbssddqs.shortlink.project.toolkit.HashUtil;
import org.swqsbssddqs.shortlink.project.toolkit.LinkUtil;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static org.swqsbssddqs.shortlink.project.common.constant.RedisCacheConstant.*;

@Service
@Slf4j
@RequiredArgsConstructor
public class ShortLinkServiceImpl extends ServiceImpl<ShortLinkMapper, ShortLinkDO> implements ShortLinkService {
    private final RBloomFilter<String> shortLinkCreateCachePenetrationBloomFilter;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final StringRedisTemplate stringRedisTemplate;
    private final RedissonClient redissonClient;

    private final ShortLinkGotoWhiteList shortLinkGotoWhiteList;
    private final ShortLinkStatsSaveProducer shortLinkStatsSaveProducer;
//    private final KafkaShortLinkStatsSaveProducer kafkaShortLinkStatsSaveProducer;

    @Value("${short-link.domain.default}")
    private String createShortLinkDefaultDomain;
    @Value("${short-link.locale.amap-api-key}")
    private String amap_api_key;
    @Value("${short-link.locale.url}")
    private String amap_api_url;


    @Override
    public ShortLinkBatchCreateRespDTO batchCreateShortLink(ShortLinkBatchCreateReqDTO requestParam) {
        List<String> originUrls = requestParam.getOriginUrls();
        List<String> describes = requestParam.getDescribes();
        List<ShortLinkBaseInfoRespDTO> result = new ArrayList<>();
        for (int i = 0; i <requestParam.getOriginUrls().size(); i++) {
            ShortLinkCreatReqDTO build = ShortLinkCreatReqDTO.builder()
                    .originUrl(originUrls.get(i))
                    .describe(describes.get(i))
                    .createdType(requestParam.getCreatedType())
                    .gid(requestParam.getGid())
                    .createdType(requestParam.getCreatedType())
                    .validDate(requestParam.getValidDate())
                    .validDate(requestParam.getValidDate())
                    .build();
           try {
               ShortLinkCreateRespDTO shortLink = createShortLink(build);
               ShortLinkBaseInfoRespDTO build1 = ShortLinkBaseInfoRespDTO.builder()
                       .describe(describes.get(i))
                       .originUrl(originUrls.get(i))
                       .fullShortUrl(shortLink.getFullShortUrl())
                       .build();
               result.add(build1);
           }catch (Throwable e){
               log.error("批量创建短链接失败，原始参数：{}", originUrls.get(i));
           }
        }
        ShortLinkBatchCreateRespDTO build = ShortLinkBatchCreateRespDTO.builder()
                .total(result.size())
                .baseLinkInfos(result)
                .build();

        return build;
    }


    @Transactional
    public ShortLinkCreateRespDTO createShortLink(ShortLinkCreatReqDTO shortLinkCreatReqDTO) {
        checkGotoWhiteList(shortLinkCreatReqDTO.getOriginUrl());
        String uri = generateShortUri(shortLinkCreatReqDTO);

        String fullShortUrl = createShortLinkDefaultDomain+"/"+uri;
        ShortLinkDO shortLinkDO = BeanUtil.copyProperties(shortLinkCreatReqDTO, ShortLinkDO.class);
        shortLinkDO.setDomain(createShortLinkDefaultDomain);
        shortLinkDO.setShortUri(uri);
        shortLinkDO.setFullShortUrl(fullShortUrl);
        shortLinkDO.setEnableStatus(1);
        shortLinkDO.setFavicon(getFaviconByUrl(shortLinkCreatReqDTO.getOriginUrl()));
        ShortLinkGotoDO shortLinkGotoDO = ShortLinkGotoDO.builder()
                .gid(shortLinkCreatReqDTO.getGid())
                .fullShortUrl(fullShortUrl).build();

        try {
            baseMapper.insert(shortLinkDO);
            shortLinkGotoMapper.insert(shortLinkGotoDO);
        }catch (DuplicateKeyException e){
            // 首先判断是否存在布隆过滤器，如果不存在直接新增
            if (!shortLinkCreateCachePenetrationBloomFilter.contains(fullShortUrl)) {
                shortLinkCreateCachePenetrationBloomFilter.add(fullShortUrl);
            }
            throw new ServiceException(String.format("短链接：%s 生成重复", uri));
        }
        stringRedisTemplate.opsForValue().set(
                String.format(SHORT_LINK_REDIRECT_KEY, fullShortUrl),
                shortLinkCreatReqDTO.getOriginUrl(),
                LinkUtil.getLinkCacheValidTime(shortLinkCreatReqDTO.getValidDate()),TimeUnit.MILLISECONDS);

        shortLinkCreateCachePenetrationBloomFilter.add(fullShortUrl);
               ShortLinkCreateRespDTO build = ShortLinkCreateRespDTO.builder()
                .gid(shortLinkDO.getGid())
                .fullShortUrl(shortLinkDO.getFullShortUrl())
                .originUrl(shortLinkDO.getOriginUrl())
                .build();
        return build;
    }
    @SneakyThrows
    private String getFaviconByUrl(String originUrl) {
        Document document = Jsoup.connect(originUrl).get();
        Element faviconLink = document.select("link[rel~=(?i)^(shortcut )?icon]").first();
        if (faviconLink != null) {
            return faviconLink.attr("abs:href");
        }
        return null;
    }

    @Override
    public IPage<ShortLinkPageRespDTO> pageShortLink(ShortLinkPageReqDTO shortLinkPageReqDTO) {
//        LambdaQueryWrapper<ShortLinkDO> eq = Wrappers.lambdaQuery(ShortLinkDO.class)
//                .eq(ShortLinkDO::getGid, shortLinkPageReqDTO.getGid())
//                .eq(ShortLinkDO::getEnableStatus, 1)
//                .eq(ShortLinkDO::getDelFlag, 0);
//        IPage<ShortLinkDO> resultPage = baseMapper.selectPage(shortLinkPageReqDTO, eq);

        IPage<ShortLinkDO> resultPage = baseMapper.selectShortLinkPage(shortLinkPageReqDTO);
        return resultPage.convert(each->{
            ShortLinkPageRespDTO result = BeanUtil.copyProperties(each, ShortLinkPageRespDTO.class);
            result.setDomain("http://" + result.getDomain());
            return result;
        } );
    }

    @Override
    public List<ShortLinkGroupCountQueryRespDTO> listGroupShortLinkCount(List<String> gidList) {
        QueryWrapper<ShortLinkDO> wrapper = Wrappers.query(new ShortLinkDO())
                .select("gid,count(*) as shortLinkCount")
                .in("gid", gidList)
                .eq("del_flag", 0)
                .eq("enable_status", 1)
                .groupBy("gid");
        List<Map<String, Object>> maps = baseMapper.selectMaps(wrapper);
        return BeanUtil.copyToList(maps, ShortLinkGroupCountQueryRespDTO.class);

    }

    @Override
    @Transactional
    public Void updateShortLink(ShortLinkUpdateReqDTO param) {
        checkGotoWhiteList(param.getOriginUrl());
        //查询originGid
        LambdaQueryWrapper<ShortLinkDO> eq = Wrappers.lambdaQuery(ShortLinkDO.class)
                .eq(ShortLinkDO::getGid, param.getOriginGid())
                .eq(ShortLinkDO::getFullShortUrl, param.getFullShortUrl())
                .eq(ShortLinkDO::getDelFlag, 0)
                .eq(ShortLinkDO::getEnableStatus, 1);
        ShortLinkDO shortLinkDO = baseMapper.selectOne(eq);
        if(shortLinkDO==null){
            throw new ClientException("短链接不存在！");
        }
        LambdaUpdateWrapper<ShortLinkDO> updateWrapper = Wrappers.lambdaUpdate(ShortLinkDO.class)
                .eq(ShortLinkDO::getFullShortUrl, param.getFullShortUrl())
                .eq(ShortLinkDO::getGid, param.getOriginGid())
                .eq(ShortLinkDO::getDelFlag, 0)
                .eq(ShortLinkDO::getEnableStatus, 1);

        if(Objects.equals(param.getOriginGid(),param.getGid())){
            updateWrapper.set(Objects.equals(param.getValidDateType(), VailDateTypeEnum.PERMANENT.getType()), ShortLinkDO::getValidDate, null);
            ShortLinkDO updateDO = ShortLinkDO.builder().originUrl(param.getOriginUrl())
                    .validDate(param.getValidDate())
                    .validDateType(param.getValidDateType())
                    .describe(param.getDescribe()).build();
            baseMapper.update(updateDO,updateWrapper);
        }else{
            RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_UPDATE_GID_KEY, param.getFullShortUrl()));
            RLock rLock = readWriteLock.writeLock();
            if(!rLock.tryLock()){
                throw new ServiceException("短链接正在被访问，请稍后再试·····");
            }
            try {
                log.info("开始睡觉");
                Thread.sleep(10000L);
                log.info("睡觉结束");
                ShortLinkDO delShortLinkDO = new ShortLinkDO();
                delShortLinkDO.setDelFlag(1);
                delShortLinkDO.setDelTime(System.currentTimeMillis());
                baseMapper.update(delShortLinkDO,updateWrapper);

                shortLinkDO.setOriginUrl(param.getOriginUrl())
                        .setId(null)
                        .setValidDate(param.getValidDate())
                        .setValidDateType(param.getValidDateType())
                        .setDescribe(param.getDescribe());
                shortLinkDO.setGid(param.getGid());
                baseMapper.insert(shortLinkDO);
                LambdaUpdateWrapper<ShortLinkGotoDO> eq1 = Wrappers.lambdaUpdate(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, param.getFullShortUrl());
                ShortLinkGotoDO shortLinkGotoDO = ShortLinkGotoDO.builder().gid(param.getGid()).build();
                shortLinkGotoMapper.update(shortLinkGotoDO,eq1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                rLock.unlock();
            }

        }
        return null;
    }

    @Override
    @SneakyThrows
    public void redirectUrl(String shortUri, ServletRequest request, ServletResponse response) {
        String serverName = request.getServerName();
        String port = Optional.of(request.getServerPort())
                .filter(each -> !Objects.isNull(each))
                .map(String::valueOf)
                .map(each -> ":" + each)
                .orElse("");
        String fullShortUrl = serverName+port+"/"+shortUri;
        String originUrl = stringRedisTemplate.opsForValue().get(String.format(SHORT_LINK_REDIRECT_KEY, fullShortUrl));
        if(StrUtil.isNotBlank(originUrl)){
            actualShortLinkStats(createShortLinkStatsRecordDTO(fullShortUrl, request, response));
            ((HttpServletResponse) response).sendRedirect(originUrl);
            return;
        }


        //布隆过滤器+存空值解决缓存穿透
        if(!shortLinkCreateCachePenetrationBloomFilter.contains(fullShortUrl)){
            ((HttpServletResponse) response).sendRedirect("/page/notfound");
            return;
        }
        originUrl = stringRedisTemplate.opsForValue().get(String.format(SHORT_LINK_REDIRECT_IS_NULL_KEY, fullShortUrl));
        if("-".equals(originUrl)){
            ((HttpServletResponse) response).sendRedirect("/page/notfound");
            return;
        }

        //双重检查锁-解决缓存击穿
        RLock lock = redissonClient.getLock(String.format(LOCK_SHORT_LINK_REDIRECT_KEY, fullShortUrl));
        lock.lock();
        try {
            //获取锁后，再检查空值和非空值
            originUrl = stringRedisTemplate.opsForValue().get(String.format(SHORT_LINK_REDIRECT_KEY, fullShortUrl));
            if(StrUtil.isNotBlank(originUrl)){
                actualShortLinkStats(createShortLinkStatsRecordDTO(fullShortUrl, request, response));
                ((HttpServletResponse) response).sendRedirect(originUrl);
                return;
            }
            originUrl = stringRedisTemplate.opsForValue().get(String.format(SHORT_LINK_REDIRECT_IS_NULL_KEY, fullShortUrl));
            if("-".equals(originUrl)){
                ((HttpServletResponse) response).sendRedirect("/page/notfound");
                return;
            }

            LambdaQueryWrapper<ShortLinkGotoDO> eq = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                    .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
            ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(eq);
            if(shortLinkGotoDO==null){
                stringRedisTemplate.opsForValue().set(String.format(SHORT_LINK_REDIRECT_IS_NULL_KEY, fullShortUrl),"-");
                ((HttpServletResponse) response).sendRedirect("/page/notfound");
                return;
            }
            LambdaQueryWrapper<ShortLinkDO> eq1 = Wrappers.lambdaQuery(ShortLinkDO.class)
                    .eq(ShortLinkDO::getGid, shortLinkGotoDO.getGid())
                    .eq(ShortLinkDO::getFullShortUrl, fullShortUrl)
                    .eq(ShortLinkDO::getDelFlag,0)
                    .eq(ShortLinkDO::getEnableStatus,1);
            ShortLinkDO shortLinkDO = baseMapper.selectOne(eq1);

            if (shortLinkDO == null || (shortLinkDO.getValidDate() != null && shortLinkDO.getValidDate().before(new Date()))) {
                stringRedisTemplate.opsForValue().set(String.format(SHORT_LINK_REDIRECT_IS_NULL_KEY, fullShortUrl), "-", 30, TimeUnit.MINUTES);
                ((HttpServletResponse) response).sendRedirect("/page/notfound");
                return;
            }
            stringRedisTemplate.opsForValue().set(
                    String.format(SHORT_LINK_REDIRECT_KEY,fullShortUrl),
                    shortLinkDO.getOriginUrl(),
                    LinkUtil.getLinkCacheValidTime(shortLinkDO.getValidDate()),
                    TimeUnit.MILLISECONDS
            );
//            shortLinKStats(fullShortUrl, request, response);
            actualShortLinkStats(createShortLinkStatsRecordDTO(fullShortUrl, request, response));
            ((HttpServletResponse) response).sendRedirect(shortLinkDO.getOriginUrl());

        }finally {
            lock.unlock();
        }
    }

    public void actualShortLinkStats(ShortLinkStatsRecordDTO param){
        //redis stream异步落库stats
        Map<String, String> producerMap = new HashMap<>();
        producerMap.put("statsRecord", JSON.toJSONString(param));
        shortLinkStatsSaveProducer.send(producerMap);

        //kafka异步落库stats
        //kafkaShortLinkStatsSaveProducer.sendStatsSaveMessage(param);


    }
    public ShortLinkStatsRecordDTO createShortLinkStatsRecordDTO(String fullShortUrl, ServletRequest request, ServletResponse response){
        HttpServletRequest req = (HttpServletRequest)request;
        HttpServletResponse resp = (HttpServletResponse) response;
        Cookie[] cookies = req.getCookies();
        //统计uv，pv
        AtomicReference<String> uv = new AtomicReference<>();
        AtomicBoolean uvFirstFlag = new AtomicBoolean();
        Runnable runnable = ()->{
            uv.set(UUID.randomUUID().toString());
            uvFirstFlag.set(true);
            Cookie cookie = new Cookie("uv",uv.get());
            cookie.setPath(req.getRequestURI());
            cookie.setMaxAge(60 * 60 * 24 * 30);
            resp.addCookie(cookie);
            stringRedisTemplate.opsForSet().add(String.format(SHORT_LINK_STATS_UV_KEY, fullShortUrl), uv.get());
        };
        if(ArrayUtil.isNotEmpty(cookies)){
            Arrays.stream(cookies)
                    .filter(each->"uv".equals(each.getName()))
                    .findFirst()
                    .map(Cookie::getValue)
                    .ifPresentOrElse(each->{
                        uv.set(each);
                        Long add = stringRedisTemplate.opsForSet().add(String.format(SHORT_LINK_STATS_UV_KEY, fullShortUrl), each);
                        uvFirstFlag.set(add!=null&&add>0);
                    },runnable);
        }else{
            runnable.run();
        }
        String remoteAddr = LinkUtil.getActualIp(((HttpServletRequest) request));
        String os = LinkUtil.getOs(((HttpServletRequest) request));
        String browser = LinkUtil.getBrowser(((HttpServletRequest) request));
        String device = LinkUtil.getDevice(((HttpServletRequest) request));
        String network = LinkUtil.getNetwork(((HttpServletRequest) request));
        Long uipAdded = stringRedisTemplate.opsForSet().add(String.format(SHORT_LINK_STATS_IPV_KEY, fullShortUrl),remoteAddr);
        boolean uipFirstFlag = uipAdded != null && uipAdded > 0L;
        return ShortLinkStatsRecordDTO.builder()
                .fullShortUrl(fullShortUrl)
                .uv(uv.get())
                .uvFirstFlag(uvFirstFlag.get())
                .uipFirstFlag(uipFirstFlag)
                .remoteAddr(remoteAddr)
                .os(os)
                .browser(browser)
                .device(device)
                .network(network)
                .currentDate(new Date())
                .build();
    }
    public String generateShortUri(ShortLinkCreatReqDTO shortLinkCreatReqDTO){
        String shortUri;
        int count =0 ;
        while (true){
            String originUrl = shortLinkCreatReqDTO.getOriginUrl()+ UUID.randomUUID();
            shortUri = HashUtil.hashToBase62(originUrl);
            if(count>10){
                throw new ServiceException("短链接频繁生成，请稍后重试！");
            }
            if(!shortLinkCreateCachePenetrationBloomFilter.contains(createShortLinkDefaultDomain+"/"+shortUri)){
                break;
            }
            count++;
        }

        return shortUri;
    }
    public void checkGotoWhiteList(String originUrl){

        Boolean enable = shortLinkGotoWhiteList.getEnable();
        if (enable == null || !enable) {
            return;
        }
        String domain = LinkUtil.extractDomain(originUrl);
        if (StrUtil.isBlank(domain)) {
            throw new ClientException("跳转链接填写错误");
        }
        List<String> details = shortLinkGotoWhiteList.getDetails();
        if (!details.contains(domain)) {
            throw new ClientException("演示环境为避免恶意攻击，请生成以下网站跳转链接：" + shortLinkGotoWhiteList.getNames());
        }
    }
}
